package com.wudl.kafka

import kafka.serializer.StringDecoder
import org.apache.spark.SparkConf
import org.apache.spark.streaming.kafka.KafkaUtils
import org.apache.spark.streaming.{Seconds, StreamingContext}

object KafkaStream {
  def main(args: Array[String]): Unit = {

    val conf = new SparkConf().setMaster("local[1]").setAppName("KafkaStream")
    val ssc = new StreamingContext(conf, Seconds(3))

    val params = Map[String, String]("bootstrap.servers" -> "node01.com:6667", "group.id" -> "1")
//    KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](
//      ssc, params, set("pgy")
//    ).flatMap {
//      case (_, v) =>
//        v.split("\\W+")
//    }
//      .map((_, 1))
//      .reduceByKey(_ + _)
//      .print()
//  }

    KafkaUtils
      .createDirectStream[String, String, StringDecoder, StringDecoder](
        ssc,
        params,
        Set("pyg")
      )
      .flatMap {
        case (_, v) =>
          v.split(",")
      }
      .map((_, 1))
      .reduceByKey(_ + _)
      .print()

    ssc.start()
    ssc.awaitTermination()
  }

}
